// -*-mode:c++; coding:utf-8-*-

// system headers
#include <signal.h>

// standard headers

#include <string>
#include <iostream>
#include <memory>
#include <exception>
#include <fstream>
#include <sstream>

#include <boost/lexical_cast.hpp>

// global lib headers
//#include <xthread/exceptions.hpp>
#include <xthread/guard.hpp>
#include <xthread/fast_lock.hpp>
#include <xthread/read_write_lock.hpp>

// local lib headers

#include <constants.hpp>
#include <protocol.hpp>
#include <xbase_exception.hpp>
#include <xbase_utility.hpp>
#include <fs.hpp>

// local headers

#include "tablet_server_version.hpp"
#include "log_thread.hpp"
#include "tablet.hpp"
#include "tablet_server_configuration.hpp"
#include "global.hpp"
#include "naming_strategy.hpp"
#include "service_provider.hpp"
#include "table_schema.hpp"
#include "table_schema_manager.hpp"
#include "table_manager.hpp"

#include "service.hpp"
#include "minor_compact_checker.hpp"
#include "major_compact_checker.hpp"
#include "scanner_checker.hpp"
#include "tablet_checker.hpp"

// the defination header at last

void init_system()
{
	global::init();

	fs::path _home = global::the_conf->home();
	fs::path _fs_conf_path = _home / "conf" / "gfsclient.conf";
	fs::init(_fs_conf_path.string());

	//
	// 开始初始化并填充数据结构
	//
	global::the_address_info->m_hostname = global::the_conf->tabletserver_ip;
	global::the_address_info->m_port = global::the_conf->client_listen_port;
	global::the_address_info->m_startcode = cur_micro_seconds();

	// other
	sstable_block_cache_manager::init_instance(global::the_conf->max_cache_memory);
}

void reset_system()
{
	LOG(WARN, "reset system");
	global::reset();

	sstable_block_cache_manager::instance().reset_instance();
}

class working_thread_manager
{
public:
	void start_all() {
		m_threads.create_thread(boost::bind(&minor_compact_checker::run,
						    &m_minor_compact_checker));
		m_threads.create_thread(boost::bind(&major_compact_checker::run,
						    &m_major_compact_checker));
		m_threads.create_thread(boost::bind(&scanner_checker::run,
						    &m_scanner_checker));
		m_threads.create_thread(boost::bind(&tablet_checker::run,
						    &m_tablet_checker));
	}

	void stop_all() {
		m_minor_compact_checker.stop();
		m_major_compact_checker.stop();
		m_scanner_checker.stop();
		m_tablet_checker.stop();
	}

	void join_all() {
		m_threads.join_all();
	}

private:
	boost::thread_group m_threads;

	minor_compact_checker m_minor_compact_checker;
	major_compact_checker m_major_compact_checker;
	scanner_checker m_scanner_checker;
	tablet_checker m_tablet_checker;
};


int main(int argc, char *argv[]) {
	// Block all signals for background thread.
	sigset_t new_mask;
	sigfillset(&new_mask);
	sigset_t old_mask;
	pthread_sigmask(SIG_BLOCK, &new_mask, &old_mask);

	init_system();

	// 创建子表服务器目录
	std::string server_dirname = naming_strategy::server_dirname(global::the_address_info->m_hostname,
								     global::the_address_info->m_port,
								     global::the_address_info->m_startcode);
	fs::path _root = global::the_conf->xbase_root_path;
	_root = _root / server_dirname;
	
	bool _ok = fs::mkdir(_root);
	if (! _ok) {
		LOG(FATAL, "create tablet server dir " << _root << " failed.");
		return -1;
	}
	global::the_tablet_server_path = _root;
	LOG(TRACE, "tablet server dir " << _root << " created.");
	
	// must here, after server path created
	global::the_log_manager->start_service();
	// and before any socket opened.

	working_thread_manager working_threads;
	working_threads.start_all();

	application app(global::the_conf->client_handler_number,
			global::the_conf->epoll_number,
			global::the_conf->tabletserver_ip,
			boost::lexical_cast<std::string>(global::the_conf->client_listen_port),
			global::the_conf->master_ip,
			boost::lexical_cast<std::string>(global::the_conf->master_port),
			global::the_conf->heart_beat_interval / 10); // TODO

	//starting_working_thread
	app.start();

	// Restore previous signals.
	pthread_sigmask(SIG_SETMASK, &old_mask, 0);

	// 检查与master的连接
	const std::size_t interval = global::the_conf->heart_beat_interval;

	const std::size_t recheck_interval = 1;
	std::size_t lost_count = 0;
	std::vector<fs::file_info> infos;

	time_t heartbeat_interval = interval >> 1; // 使第一次心跳提早一点发
	time_t last_heartbeat_time = cur_seconds();
	while(true)
	{
		if(global::stop_request)
		{
			LOG(WARN, "stop requested, report to master and quit");
			global::the_service->unload_all_tablet("tabletserver: stop requested, so unload all tablet and quit");
			global::the_service->report_tablet_server_quit();
			break;
		}

		const time_t cur_time = cur_seconds();
		const time_t send_interval = cur_time - last_heartbeat_time;
		if(send_interval < heartbeat_interval)
		{
			sleep(heartbeat_interval - send_interval);
		}
		
		// send
		LOG(TRACE, "it's heartbeat time");
		response::pointer msg(response::factory_type::construct());
		global::the_service->construct_heartbeat(msg);
		const int ret = global::the_master_session->heartbeat(msg);
		LOG_IF_ELSE((ret < 0),
			    WARN, "heartbeat error",
			    TRACE, "heartbeat done");

		// update
		last_heartbeat_time = cur_seconds();
		// 发送失败，则设置为3秒后重试
		heartbeat_interval = (ret < 0) ? 3 : interval;

		// 检查目录如此复杂是因为GFS下有时候操作会失败，这样避
		// 免了因为失败而导致误认为目录不存在的可能。
		//
		// TODO: 加入对error_no的检查，仅当error_no也显示为目录
		// 不存在，才认为目录真的不存在
		bool no_exists = (! fs::exists(global::the_tablet_server_path)); // 第一次检查
		if(no_exists)
		{
			sleep(recheck_interval); // wait a moment
			no_exists = (! fs::list_files(infos, global::the_tablet_server_path)); // 第二次检查
			infos.clear();
		}
		if(no_exists)
		{
			if(lost_count >= 3) // 连续3遍以上检查都失败，才认为失败。
			{
				LOG(FATAL, "server dir didn't exists,"
				    << " so quit and reset the server.");
				break;
			}
			++ lost_count;
		}
		else
		{
			lost_count = 0; // reset
		}
	}

	// // Wait for signal indicating time to shut down.
	// sigset_t wait_mask;
	// sigemptyset(&wait_mask);
	// sigaddset(&wait_mask, SIGINT);
	// sigaddset(&wait_mask, SIGQUIT);
	// sigaddset(&wait_mask, SIGTERM);
	// pthread_sigmask(SIG_BLOCK, &wait_mask, 0);
	// int sig = 0;
	// sigwait(&wait_mask, &sig);

	// 先将网络处理停掉
	app.stop();
	app.join_all();

	// 再将工作线程停掉
	working_threads.stop_all();
	working_threads.join_all();

	global::the_master_session.reset();
	// master_session中用到了app中的io_service，所以需要先在此释放
	// 掉。否则app将先于master_session析构。

	reset_system();

	return 0;
}
